[improve][ml] Pin ledger callbacks to the managed ledger thread with withOrderingKey - #26599

…withOrderingKey

A managed ledger runs on bookKeeper.getMainWorkerPool().chooseThread(name),
while the BookKeeper client pins each LedgerHandle to chooseThread(ledgerId)
of the same pool. The two keys hash to different threads, so every add
completion pays an extra hop from the ledger thread to the managed ledger
thread, and each bookie response wakes a thread that is not the one doing
the ledger's work.

BookKeeper 4.18.1 added CreateBuilder/OpenBuilder.withOrderingKey
(apache/bookkeeper#4881): the key is resolved with chooseThread(key) on
the client's main worker pool, so passing the managed ledger name makes
every callback of the handle run on the managed ledger thread. This
passes the name at asyncCreateLedger (data and cursor ledgers) and at
every open in ManagedLedgerImpl, ReadOnlyManagedLedgerImpl,
ShadowManagedLedgerImpl and ManagedCursorImpl. The offline-stats opens
in ManagedLedgerFactoryImpl have no managed ledger thread and are left
unkeyed.

ManagedLedgerBkTest gains a test against a real bookie that checks the
read callbacks of the current ledger, of a closed ledger re-opened
through getLedgerHandle, and of the cursor ledger all run on the managed
ledger's executor thread.

…dering-key

…d ledger thread

With the ledger callbacks pinned to the managed ledger thread through
withOrderingKey, OpAddEntry.addComplete, PendingReadsManager.attach and
OpReadEntry.complete no longer need to re-queue their processing on the
executor: ThreadBoundExecutor.executeOrRun runs it inline when already
on that thread and falls back to execute() otherwise (mock clients still
deliver callbacks from their own thread).

The managed ledger executor is now typed as ThreadBoundExecutor, cast
from the main worker pool thread the same way the BookKeeper client does
for its ledger handles. Tests that stub getExecutor() hand out a
ThreadBoundExecutor: a mock, a worker thread of the OrderedExecutor the
mock client uses, or the new DirectThreadBoundExecutor for the test that
needs same-thread completion.

… writes on the mock read view

Review follow-ups:

- The builder-based opens bridge their future to the legacy OpenCallback
  with whenComplete and dropped the returned stage, so anything the
  callback threw (for example a RejectedExecutionException when the
  managed ledger executor is already shut down) was swallowed instead of
  being logged the way BookKeeper's executor did. All six sites now go
  through ManagedLedgerImpl.completeOpenCallback, which logs it.
- PulsarMockBookKeeper.newOpenLedgerOp() used thenComposeAsync, which
  short-circuits on an already failed programmed-failure future and
  completed the open inline on the caller's thread. It now always
  completes on the mock executor, as the legacy mock open path did.
- PulsarMockReadHandle rejects writes with IllegalOpException, like
  ReadOnlyLedgerHandle, instead of falling through to LedgerHandle's
  write path and the mock's null client internals.

…dering-key

# Conflicts:
#	managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
#	managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ShadowManagedLedgerImpl.java

…dering-key

# Conflicts:
#	managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
#	managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
#	managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ShadowManagedLedgerImpl.java

OpReadEntry.complete() goes back to running inline on the ledger thread,
which is the point of the ordering key: a read completion should not pay
a queue hop. The recursion that the queue used to bound (a fully cached
read completes synchronously and OpScan or the replicator issue the next
read from the callback) is bounded by a nesting depth instead: past
MAX_NESTED_INLINE_COMPLETIONS inline completions the next one is queued,
which unwinds the stack, and the chain resumes inline from the executor
loop. ManagedLedgerImpl carries the depth counter; it is only touched
from the ledger thread.

The scan regression test now asserts that the stack depth seen by the
last entries stays within that bound instead of staying flat.

The nesting cap for inline read completions is a property of the
thread's stack, not of one managed ledger: several independent reads
interleave on a ledger thread without nesting, and a thread serves
several managed ledgers. Keep the depth in a FastThreadLocal in
OpReadEntry instead of a field on ManagedLedgerImpl, so only actual
nesting on the current thread counts toward the cap.

…dering-key

# Conflicts:
#	managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java